package net.sf.calibur.transport.mina;

import static java.lang.String.*;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;

import net.sf.calibur.jmx.ManagedBean;
import net.sf.calibur.transport.Session;
import net.sf.calibur.transport.SessionManager;
import net.sf.calibur.transport.Transporter;
import net.sf.calibur.transport.message.Message;

import org.apache.mina.core.buffer.IoBuffer;
import org.apache.mina.core.filterchain.DefaultIoFilterChainBuilder;
import org.apache.mina.core.service.IoHandlerAdapter;
import org.apache.mina.core.session.IdleStatus;
import org.apache.mina.core.session.IoSession;
import org.apache.mina.filter.codec.ProtocolCodecFilter;
import org.apache.mina.filter.executor.ExecutorFilter;
import org.apache.mina.filter.logging.LoggingFilter;
import org.apache.mina.transport.socket.SocketSessionConfig;
import org.apache.mina.transport.socket.nio.NioSocketAcceptor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.softee.management.annotation.MBean;
import org.softee.management.annotation.ManagedAttribute;

@MBean
public class MinaTcpTransporter implements Transporter, ManagedBean {
	public final static String KEY_IOBUFFER = "IoBuffer";

	private String name;
	private InetSocketAddress listenAddress;
	private SessionManager sessionManager;
	private NioSocketAcceptor acceptor;
	private Logger logger = LoggerFactory.getLogger(getClass());

	/**
	 * 发送数据时重用IoBuffer对象的池，由getSendBuffer()获得一个实例，在IoHandlerAdapter#
	 * messageSent方法中回收。
	 */
	private Queue<IoBuffer> sendBufferPool;

	public MinaTcpTransporter(String name, SocketAddress addr,
			SessionManager sessionManager) {
		this.name = name;
		this.listenAddress = (InetSocketAddress) addr;
		this.sessionManager = sessionManager;
		this.sendBufferPool = new ConcurrentLinkedQueue<IoBuffer>();
	}

	@ManagedAttribute
	@Override
	public String getName() {
		return name;
	}

	@ManagedAttribute
	public int getSendBufferPoolSize() {
		return sendBufferPool.size();
	}

	/**
	 * 获得一个发送数据时使用的IoBuffer实例。
	 * 
	 * @return
	 */
	public IoBuffer getSendBuffer() {
		IoBuffer buffer = sendBufferPool.poll();
		if (buffer == null) {
			buffer = IoBuffer.allocate(1024, true);
			buffer.setAutoExpand(true);
		}

		buffer.clear();
		return buffer;
	}

	@Override
	public void start() throws Exception {
		acceptor = new NioSocketAcceptor(Runtime.getRuntime()
				.availableProcessors() + 1);
		DefaultIoFilterChainBuilder chain = acceptor.getFilterChain();

		// 增加Logger
		if (logger.isDebugEnabled()) {
			chain.addLast("logger", new LoggingFilter());
		}

		// 设置encoder/decoder
		chain.addLast("codec", new ProtocolCodecFilter(
				new MessageCodecFactory()));

		// 设置ExecutorFilter(MINA 2.0开始要求)
		chain.addLast("executor", new ExecutorFilter());

		// 设置session的handler
		acceptor.setHandler(new IoHandlerAdapter() {
			private Session getCaliburSession(IoSession session) {
				return (Session) session
						.getAttribute(SessionManager.ATTR_SESSION);
			}

			private void closeIoSession(IoSession session) {
				try {
					session.close(true);
				} catch (Throwable t) {
					logger.warn("Close IoSession failed.", t);
				}
			}

			@Override
			public void sessionCreated(IoSession session) throws Exception {
				super.sessionCreated(session);

				Session caliburSession = new MinaSessionImpl(
						MinaTcpTransporter.this, session, sessionManager);
				session.setAttribute(SessionManager.ATTR_SESSION,
						caliburSession);
				sessionManager.sessionConnected(caliburSession);
			}

			@Override
			public void sessionClosed(IoSession session) throws Exception {
				Session caliburSession = getCaliburSession(session);
				if (caliburSession != null) {
					sessionManager.sessionClosed(caliburSession);
				}
			}

			@Override
			public void exceptionCaught(IoSession session, Throwable cause)
					throws Exception {
				try {
					Session caliburSession = getCaliburSession(session);
					sessionManager.exceptionCaught(caliburSession, cause);
				} catch (Throwable t) {
					logger.error("!!!Transport caught exception.", cause);
					logger.error(
							"!!!But exception process occured exception again.",
							t);
				}
			}

			@Override
			public void messageReceived(IoSession session, Object message)
					throws Exception {
				Session caliburSession = getCaliburSession(session);
				if (caliburSession != null) {
					Message msg = (Message) message;
					sessionManager.messageReceived(caliburSession, msg);
				} else {
					closeIoSession(session);
				}
			}

			@Override
			public void sessionIdle(IoSession session, IdleStatus status)
					throws Exception {
				Session caliburSession = getCaliburSession(session);
				if (caliburSession != null) {
					sessionManager.sessionIdle(caliburSession);
				} else {
					closeIoSession(session);
				}
			}

			@Override
			public void messageSent(IoSession session, Object message)
					throws Exception {
				if (message instanceof IoBuffer) {
					// 把IoBuffer实例归还到sendBufferPool中
					sendBufferPool.add((IoBuffer) message);
				}
			}

			@Override
			public void sessionOpened(IoSession session) throws Exception {
				super.sessionOpened(session);
			}
		});

		// 设置read buffer size
		SocketSessionConfig config = acceptor.getSessionConfig();
		config.setReadBufferSize(8192);
		config.setSendBufferSize(8192);
		config.setTcpNoDelay(true);
		config.setKeepAlive(true);


		// 设置读空闲时间，一个session如果超过这个时间没有收到客户端发来的任何数据，
		// 则会调用handler的sessionIdle方法，用以实现心跳检测功能。
		config.setIdleTime(IdleStatus.READER_IDLE, 60);

		// 绑定地址并开始监听端口
		acceptor.setReuseAddress(true);
		acceptor.bind(listenAddress);
		logger.info(format("Transporter %s listened on %s.", getName(),
				listenAddress.toString()));
	}

	@Override
	public void shutdown() {
		acceptor.unbind();
		for (IoSession session : acceptor.getManagedSessions().values()) {
			session.close(true);
		}
	}

	@Override
	public long getActivationTime() {
		return acceptor.getActivationTime();
	}

	@ManagedAttribute
	public long getAliveTime() {
		return System.currentTimeMillis() - acceptor.getActivationTime();
	}

	@ManagedAttribute
	@Override
	public int getManagedSessionCount() {
		return acceptor.getManagedSessionCount();
	}

	@ManagedAttribute
	@Override
	public long getRxBytes() {
		if (acceptor == null) {
			return 0;
		}
		return acceptor.getStatistics().getReadBytes();
	}

	@ManagedAttribute
	@Override
	public long getTxBytes() {
		if (acceptor == null) {
			return 0;
		}
		return acceptor.getStatistics().getWrittenBytes();
	}

	@Override
	public String getObjectName() {
		String objectName = new StringBuffer("calibur.transport:type=TCP,name=")
				.append(getName()).append('-')
				.append(listenAddress.getHostName()).append('_')
				.append(listenAddress.getPort()).toString();
		return objectName;
	}
}
